#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>          /* See NOTES */
#include <sys/socket.h>
#include <poll.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <string.h>
#include <pthread.h>


#define NQ     3      // number of queues
#define MAXMSZ 512    // maximum message size
#define KEY    0x123  // key for list message


// struct pollfd {
//     int   fd;         /* file descriptor */
//     short events;     /* requested events */
//     short revents;    /* returned events */
// };


/* 传给线程的参数类型 */
struct threadinfo {
    int qid; // 队列id
    int fd;  // socket fd
};

struct mymesg {
    long type; // 消息类型
    char text[MAXMSZ]; // 消息正文 (隐含了消息长度信息 MAXMSZ)
};

static void err_sys(const char* str) {
    perror(str);
    exit(1);
}

/** 
 * 线程负责从固定qid的消息队列阻塞读取消息, 如果读取到消息则通过fd-pipe写入
 */
void *helper(void *arg) {
    int n;
    struct mymesg m;
    struct threadinfo *tinfop = arg;

    for ( ; ;) {
        memset(&m, 0, sizeof (m));

        // 从消息队列阻塞读取第一个消息, 长度超过指定值则截断
        if ((n = msgrcv(tinfop->qid, &m, MAXMSZ, 0, MSG_NOERROR)) < 0) {
            err_sys("msgrcv erro");
        }

        // 向fd管道写入收到的消息正文
        if (write(tinfop->fd, m.text, n) < 0) {
            err_sys("write error");
        }
    }

}


int main() {
    int i, n, err;
    int fd[2]; // socketpair创建的双套接字对应sock fd
    int qid[NQ]; // 队列id数组
    struct pollfd pfd[NQ]; // poll监听数组
    struct threadinfo tinfo[NQ]; // 线程信息数组, 作为线程函数实参
    pthread_t tid[NQ]; // 线程id数组
    char buf[MAXMSZ]; // 缓冲区

    for (i = 0; i < NQ; ++i) {
        // 创建消息队列, 并存入数组qid
        if ((qid[i] = msgget((KEY + i), IPC_CREAT | 0666)) < 0) {
            err_sys("msg error");
        }

        printf("queue ID %d is %d\n", i, qid[i]);

        // 创建一对无名, 连接socket fd
        if (socketpair(AF_UNIX, SOCK_DGRAM, 0, fd) < 0) {
            err_sys("socketpair failed");           
        }

        // 设置poll监控的fd集
        pfd[i].fd = fd[0]; // 读端fd放入poll监控集
        pfd[i].events = POLLIN; 

        tinfo[i].qid = qid[i];
        tinfo[i].fd = fd[1]; // 写端fd传送给线程
        
        if ((err = pthread_create(&tid[i], NULL, helper, &tinfo[i])) != 0) {
            fprintf(stderr, "pthread_create error");
            exit(1);
        }
    }
    
    // 主线程poll监听读fd集, 新建的线程tid, 阻塞读取消息队列消息
    for (; ;) {
        if (poll(pfd, NQ, -1) < 0) {
            err_sys("poll error");
        }
        
        for (i = 0; i < NQ; ++i) {
            if (pfd[i].revents == pfd[i].events) { // 监听事件发生
                if ((n = read(pfd[i].fd, buf, sizeof (buf))) < 0) {
                    err_sys("read error");
                }

                buf[n] = 0; // 末尾添加null终结符
                printf("queue id %d, message %s\n", qid[i], buf);

            }
        }

    }
    
    return 0;
}

